package com.ntd.study.protocol.mqtt;

import org.eclipse.paho.client.mqttv3.*;

import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLSocketFactory;
import javax.net.ssl.TrustManagerFactory;
import java.io.FileInputStream;
import java.nio.charset.StandardCharsets;
import java.security.KeyStore;
import java.security.cert.CertificateFactory;
import java.security.cert.X509Certificate;
import java.text.SimpleDateFormat;

/**
 *
 * 功能描述：MQTT测试
 * 创建人： mao2080@sina.com
 * 创建时间：2017年7月4日 下午5:08:59
 * 修改人： mao2080@sina.com
 * 修改时间：2017年7月4日 下午5:08:59
 */
public class MQTTTest_wss {

    /**MQTT服务端ip及端口*/
    private static String host = "wss://emqxtest.fxo2opt.com:443";

    /**账号*/
    private static String username = "hjlyemqx";

    /**密码*/
    private static String password = "hjly@m@JFlEjgzBTu^ZXB";

    /**订阅的主题*/
    private static String subTopic = "a/b/c";

    /**clientID*/
    private static String clientId = "hjlyemqx";

    /**发布的主题*/
    private static String pubTopic = "a/b/c";

    /**MQTT-Client*/
    private static MqttClient client;

    /**证书路径*/
    private static String caPath = "D:\\work\\cert\\mqtt_192.168.100.222_ca.pem";

    /**
     * @throws InterruptedException
     * @throws MqttException */
    public static void main(String[] args) throws InterruptedException, MqttException {

        // 订阅消息的方法
        subscribe();
//
        publish();
    }

    /**
     *
     * 描述：订阅信息
     * @author mao2080@sina.com
     * @created 2017年7月4日 下午4:53:47
     * @since
     * @return
     */
    public static void subscribe() {
        try {
            // 创建MqttClient
            MQTTTest_wss.getClient().setCallback(new MqttCallback() {

                public void connectionLost(Throwable arg0) {

                }

                public void messageArrived(String topic, MqttMessage message) throws Exception {
                    System.out.println("MQTT Rece:" + message.toString());
                }

                public void deliveryComplete(IMqttDeliveryToken token) {

                }

            });
            MQTTTest_wss.getClient().subscribe(subTopic, 0);
            System.out.println("连接状态:" + client.isConnected());
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     *
     * 描述：获取MqttClient
     * @author mao2080@sina.com
     * @created 2017年7月6日 上午9:56:37
     * @since
     * @return
     * @throws MqttException
     */
    public static MqttClient getClient() throws MqttException{
        try {
            if(client == null){
                client = new MqttClient(host, clientId);
                MqttConnectOptions conOptions = new MqttConnectOptions();
                conOptions.setUserName(username);
                conOptions.setPassword(password.toCharArray());
                conOptions.setCleanSession(true);
                conOptions.setSocketFactory(getSSLSocktet(caPath));
                client.connect(conOptions);
            }
            if(!client.isConnected()){
                client.reconnect();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return client;
    }

    /**
     *
     * 描述：发布信息
     * @author mao2080@sina.com
     * @throws MqttException
     * @created 2017年7月4日 下午4:53:32
     * @since
     */
    public static void publish() throws MqttException {
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        try {
            MqttTopic topic = MQTTTest_wss.getClient().getTopic(pubTopic);
            MqttMessage message = new MqttMessage("测试消息发送".getBytes(StandardCharsets.UTF_8));
            message.setQos(0);
            topic.publish(message);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 获取SSLSocketFactory
     * @param caPath
     * @return
     * @throws Exception
     */
    public static SSLSocketFactory getSSLSocktet(String caPath) throws Exception {
        CertificateFactory cAf = CertificateFactory.getInstance("X.509");
        FileInputStream caIn = new FileInputStream(caPath);
        X509Certificate ca = (X509Certificate) cAf.generateCertificate(caIn);
        KeyStore caks = KeyStore.getInstance(KeyStore.getDefaultType());
        caks.load(null, null);
        caks.setCertificateEntry("ca-certificate", ca);
        TrustManagerFactory tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
        tmf.init(caks);
        caIn.close();
        SSLContext context = SSLContext.getInstance("TLSv1");
        context.init(null, tmf.getTrustManagers(), null);
        return context.getSocketFactory();
    }

}